package org.hadoop.hbase;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
public class HbaseAgg {
    private Connection con;
    private Configuration conf;
    @Before
    public void before() throws Exception {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.zookeeper.quorum", "server201");
        con = ConnectionFactory.createConnection(conf);
    }
    @After
    public void after() throws Exception {
        con.close();
    }
    /**
     * 统计行数
     */
    @Test
    public void testCount() throws Throwable {
        if (con != null) {
            AggregationClient aggregationClient = new AggregationClient(conf);
            Scan scan = new Scan();
            Long count = aggregationClient.rowCount(TableName.valueOf("ns1:stud"),
                    new LongColumnInterpreter(), scan);
            System.out.println("行数：" + count);
            aggregationClient.close();
            con.close();
        } else {
            System.out.println("连接失败");
        }
    }
    /**
     * 需要注意的是，必须要写入xx字段为Long类型才可以进行统计，否则总是返回null值
     * 如果是在命令行是，请使用stud.incr 'ROWKEY','f:xx',89的方式写入，直接就是Long类型
     */
    @Test
    public void testMax() throws Throwable {
        if (con != null) {
            AggregationClient aggregationClient = new AggregationClient(conf);
            Scan scan = new Scan();
            scan.addColumn(Bytes.toBytes("f"), Bytes.toBytes("height"));
            TableName table = TableName.valueOf("ns1:stud");
            LongColumnInterpreter columnInterpreter = new LongColumnInterpreter();
            Long max = aggregationClient.max(table, columnInterpreter, scan);
            System.out.println("Max：" + max);
            aggregationClient.close();
            con.close();
        } else {
            System.out.println("连接失败");
        }
    }
    /**
     * 需要注意的是，必须要写入xx字段为Long类型才可以进行统计，否则总是返回null值
     * 如果是在命令行是，请使用stud.incr 'ROWKEY','f:age',89的方式写入，直接就是Long类型
     */
    @Test
    public void testAvg() throws Throwable {
        if (con != null) {
            AggregationClient aggregationClient = new AggregationClient(conf);
            Scan scan = new Scan();
            scan.addColumn(Bytes.toBytes("f"), Bytes.toBytes("height"));
            TableName table = TableName.valueOf("ns1:stud");
            LongColumnInterpreter columnInterpreter = new LongColumnInterpreter();
            Double avg = aggregationClient.avg(table, columnInterpreter, scan);
            System.out.println("avg is：" + avg);//avg is：83.0
            aggregationClient.close();
            con.close();
        } else {
            System.out.println("连接失败");
        }
    }
    /**
     * 添加查询条件
     * 添加一些条件，以下为：f:name=xx的条件
     */
    @Test
    public void testSumWithFilter() throws Throwable {
        if (con != null) {
            AggregationClient aggregationClient = new AggregationClient(conf);
            Scan scan = new Scan();
            scan.addColumn(Bytes.toBytes("f"), Bytes.toBytes("height"));
            SingleColumnValueFilter filter =
                    new SingleColumnValueFilter(Bytes.toBytes("f"), Bytes.toBytes("name"),
                            CompareOperator.EQUAL, Bytes.toBytes("Jack"));
            scan.setFilter(filter);
            TableName table = TableName.valueOf("ns1:stud");
            LongColumnInterpreter columnInterpreter = new LongColumnInterpreter();
            Double avg = aggregationClient.avg(table, columnInterpreter, scan);
            System.out.println("===>avg is：" + avg);//
            aggregationClient.close();
            con.close();
        } else {
            System.out.println("连接失败..");
        }
    }
}
